package com.atguigu.flink.sql.wordcount;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * Created by 黄凯 on 2023/6/25 0025 18:00
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 程序架构:
 * *   1. 准备表的环境
 * *      1.1  流环境转表环境  StreamTableEnvironment
 * *      1.2  直接创建表环境  TableEnvironment
 * *   2. 创建表
 * *      2.1 将流转换成表
 * *      2.2 直接创建表(连接器表)
 * *   3. 表的操作
 * *      3.1 TableAPI
 * *      3.2 SQL
 * *   4. 表结果的输出
 * *      4.1 将表转换成流输出
 * *      4.2 连接器表
 * *
 * * 流表转换
 * *   1. 流转表
 * *   2. 表转流
 */
public class Flink03_BaseTableAPI {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        //1. 创建表环境
        // 1.1 流环境转表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1.2 直接创建表环境
        TableEnvironment tableEnvironment = TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());

        //2. 创建表
        // 2.1 流转表
        Table table = tableEnv.fromDataStream(stream);

        //2.2 连接器表

        //3. 表的操作
        //3.1 TableAPI
        //table.where().filter().select()

        //3.2 SQL
        //tableEnv.sqlQuery("sql")
        Table result = tableEnv.sqlQuery("select user, url, ts from " + table + " where user <> 'Tom'");

        //4.表结果输出
        // 4.1 将表再转换成流输出
        DataStream<Row> newStream = tableEnv.toDataStream(result);

//        newStream.print();

        // 4.2 连接器表
        //tableEnv.executeSql("insert into sinkTable select * from sourceTable")

        // 流表转换
        //Table newTable = tableEnv.sqlQuery("select user ,url ,ts from " + table);
        Table newTable = tableEnv.sqlQuery("select user, count(*) as cnt from " + table + " group by user");

        //将表转成流
        //tableEnv.toDataStream(newTable).print();

        //Table sink '*anonymous_datastream_sink$3*' doesn't support consuming update changes which is produced by node GroupAggregate(groupBy=[user], select=[user, COUNT(*) AS cnt])
        tableEnv.toChangelogStream(newTable).print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
